package net.wlm.realtime.funcation;


import ch.hsr.geohash.GeoHash;
import net.wlm.realtime.bean.VehicleData;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

public class HeatMapProcessAllWindowFunction extends ProcessAllWindowFunction<VehicleData,String, TimeWindow> {

    // Geohash精度（字符长度越长精度越高，通常6-8位足够）
    private static final int GEOHASH_PRECISION = 6;
    // 状态管理
    private transient MapState<String, Integer> heatMapState;
    @Override
    public void open(Configuration parameters) {
        MapStateDescriptor<String, Integer> descriptor =
                new MapStateDescriptor<>("heatMap", String.class, Integer.class);
        heatMapState = getRuntimeContext().getMapState(descriptor);
    }


    @Override
    public void process(ProcessAllWindowFunction<VehicleData, String, TimeWindow>.Context context,
                        Iterable<VehicleData> iter , Collector<String> out) throws Exception {
        // 每次窗口触发时清空状态
        heatMapState.clear();
        for (VehicleData data : iter) {
            // 1. 校验经纬度有效性
            if (!isValidCoordinate(data.getLongitude(), data.getLatitude())) {
                continue;
            }
            // 2. 获取经纬度
                double lng = data.getLongitude();
                double lat = data.getLatitude();
            // 3. 计算Geohash
            String geohash = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, GEOHASH_PRECISION);
            // 4. 统计每个Geohash区域的车辆数
            heatMapState.put(geohash, heatMapState.contains(geohash) ?
                    heatMapState.get(geohash) + 1 : 1);
        }
        // 输出
        for (Map.Entry<String, Integer> entry : heatMapState.entries()) {
            out.collect(entry.getKey()+","+entry.getValue());
        }
    }
    // 经纬度有效性校验
    private boolean isValidCoordinate(double lng, double lat) {
        return lng >= -180 && lng <= 180 &&
                lat >= -90 && lat <= 90 &&
                lng != 0 && lat != 0; // 排除零值
    }
}